
#include <readline/readline.h>

#include <iostream>

#include <boost/thread.hpp>
#include <boost/asio.hpp>
#include <boost/tokenizer.hpp>

#include <xbase_utility.hpp>
#include <tablet_name_parser.hpp>
#include <logger.hpp>

#include "io_service_pool_impl.hpp"
#include "table_info_cache.hpp"
#include "multi_row_mutator_on_tablet.hpp"
#include "session_manager.hpp"
#include "protocol.hpp"
#include "row_mutation.hpp"

using namespace xbase;

boost::mutex m;
std::size_t pending_ops;
std::deque<std::string> complete_info;

io_service_pool_ptr s_pool;
session_manager *manager;

typedef std::vector<row_mutation_data_ptr> row_vec;
typedef boost::shared_ptr<row_vec> row_vec_ptr;

void on_apply(row_vec_ptr data,
	      uint64_t start_ts,
	      std::size_t success_row,
	      const boost::system::error_code &ec)
{
	uint64_t end_ts = cur_micro_seconds();
	double used = (end_ts - start_ts);
	used /= (1000.0 * 1000.0);
	std::ostringstream out;
	out << "---" << std::endl;
	out << "[" << dump_timestamp(start_ts)
	    << " - " << dump_timestamp(end_ts)
	    << "] used " << used << " seconds" << std::endl;
	out << "written rows ";
	for(row_vec::iterator it=data->begin(); it != data->end(); ++it)
	{
		out << "\"" << (*it)->row_name() << "\" ";
	}
	out << std::endl << ", success_row = " << success_row
	    << ", error_code: "
	    << ec << " - " << ec.message() << std::endl;
	boost::mutex::scoped_lock lock(m);
	complete_info.push_back(out.str());
}

void on_connected(std::string tablet_name,
		  row_vec_ptr data,
		  uint64_t timestamp,
		  session_ptr s,
		  const boost::system::error_code &ec)
{
	if(ec)
	{
		std::ostringstream out;
		out << "connect failed: " << ec << " - " << ec.message()
		    << ", in writting rows ";
		for(row_vec::iterator it=data->begin(); it != data->end(); ++it)
		{
			out << "\"" << (*it)->row_name() << "\" ";
		}
		out << std::endl;
		out << "---" << std::endl;
		boost::mutex::scoped_lock lock(m);
		complete_info.push_back(out.str());
		return;
	}
	uint64_t start_ts = cur_micro_seconds();
	async_apply_on_tablet(*s_pool, s, tablet_name, *data, timestamp,
			      boost::bind(on_apply, data,
					  start_ts, _1, _2));
}

void show_data(const row_mutation_data &data)
{
	std::cout << data.to_string() << std::endl;
}

void command_error() 
{
	std::cout << "error: bad command." << std::endl
		  << "command list:" << std::endl
		  << "\tadd <row> <column> [value] [timestamp]" << std::endl
		  << "\tapply [row] [timestamp]" << std::endl
		  << "\tquit" << std::endl;
}

template<typename F>
void start_xbase(F f);

void user_main(int argc, char **argv);

boost::asio::ip::tcp::endpoint master_ep;
boost::asio::ip::tcp::endpoint ts_ep;
std::string tablet_name;
table_info_cache * the_table_info_cache = NULL;

int main(int argc, char **argv)
{
	if(argc != 6)
	{
		std::cout << "test_multi_row_mutator_on_tablet <master ip> <master port> <ip> <port> <tablet>" << std::endl;
		return 0;
	}
	
	boost::asio::ip::tcp::endpoint mep(
		boost::asio::ip::address::from_string(argv[1]),
		::atoi(argv[2])
		);

	boost::asio::ip::tcp::endpoint ep(
		boost::asio::ip::address::from_string(argv[3]),
		::atoi(argv[4])
		);
	
	master_ep = mep;
	ts_ep = ep;
	tablet_name = argv[5];

	start_xbase(boost::bind(user_main, argc, argv));
	return 0;
}

template<typename F>
void start_xbase(F f)
{
	// init
	
	LOG_CONFIG("../../conf/log4cplus.tablet_server.conf");

	async_module_stats::enable();

		io_service_pool_ptr pool(new impl::single_io_service());
		client_options options;
		pool->set_options(options);
		s_pool = pool;
		pool->run();
		session_manager sm(*pool);
		table_info_cache tic(*pool, master_ep, sm);
		manager = &sm;
		the_table_info_cache = &tic;
		
		// -------
		
		f();
		
		// -------
		
		// close
		
		sm.close();
		pool->stop();
		pool->join();
		
	async_module_stats::print(std::cout);
	std::cout << std::endl;
}


void user_main(int argc, char **argv)
{
	protocol::table_info_ptr table_info;

	{
		const tablet_name_parser parser(tablet_name);
		table_info_cache::sync_handler_type sh;
		std::cout << "query table info for " << parser.table_name() << " ... " << std::endl;
		the_table_info_cache->async_get_table_info(parser.table_name(),
							   sh.make_handler());
//							   boost::bind(&table_info_cache_sync_handler::operator(),
//								       &sh, _1, _2));
		sh.wait();
		if(sh.get_error_code())
		{
			std::cout << "error: " << sh.get_error_code() << " - " << sh.get_error_code().message() << std::endl;
			return;
		}
		table_info = sh.m_arg1;
		std::cout << "got table info: " << table_info->to_string() << std::endl;
	}
	
	typedef boost::tokenizer<boost::escaped_list_separator<char> > tok;
	boost::escaped_list_separator<char> separator("\\", ", ", "\"");
				
	// command
	// add "row" "group" "column" "value" timestamp
	// apply "row" timestamp
	typedef std::map<std::string, row_mutation> row_map;
	row_map rows;
	while(true)
	{
		{
			boost::mutex::scoped_lock lock(m);
			while(! complete_info.empty())
			{
				std::cout << complete_info.front();
				complete_info.pop_front();
				--pending_ops;
			}
		}

		char *input = readline("> ");
			
		if(input == NULL)
		{
			break;
		}
		const std::string copy = input;
		free(input);
		if((copy.empty()) ||
		   (copy[0] == '#')) // 注释
		{
			continue;
		}
		
		tok t(copy, separator);

		// for(tok::iterator it = t.begin();
		//     it != t.end(); ++it)
		// 	std::cout << "\"" << *it << "\",";
		// std::cout << std::endl;

		tok::iterator it = t.begin();
		if(it == t.end())
		{
			;
		}
		else if(*it == "quit")
		{
			break;
		}
		else if(*it == "add")
		{
			std::string row;
			// std::string group;
			std::string column;
			std::string value;
			std::string timestamp;

			++it;
			if(it != t.end())
			{
				row = *it;
				++it;
			}
			// if(it != t.end())
			// {
			// 	group = *it;
			// 	++it;
			// }
			if(it != t.end())
			{
				column = *it;
			}
			// 上面是必写的
			if(it == t.end())
			{
				command_error();
				continue;
			}
			++it;
			if(it != t.end())
			{
				value = *it;
				++it;
			}
			// 这是可选的
			if(it != t.end())
			{
				timestamp = *it;
			}

			if(row.empty())
			{
				std::cout << "error: " << error::bad_row_name << std::endl;
				continue;
			}

			row_map::iterator it = rows.find(row);
			if(it == rows.end())
			{
				std::pair<row_map::iterator, bool> ret =
					rows.insert(row_map::value_type(row, row_mutation(table_info, row)));
				assert(ret.second == true);
				it = ret.first;
			}
			row_mutation &mutation = it->second;
			
			uint64_t ts;
			try {
				ts = boost::lexical_cast<uint64_t>(timestamp);
			} catch(const std::exception &e) {
				ts = 0;
			}
			boost::system::error_code ec =
				mutation.add(column, value, ts);
			if(ec)
			{
				std::cout << "add failed: " << ec << " - " << ec.message() << std::endl;
			}
		}
		else if(*it == "apply")
		{
			std::string row;
			std::string timestamp;

			// 都是可选的
			++it;
			if(it != t.end())
			{
				row = *it;
				++it;
			}
			if(it != t.end())
			{
				timestamp = *it;
			}

			row_vec_ptr ptr_q(new row_vec());

			if(row.empty()) // apply all row
			{
				for(row_map::iterator it = rows.begin();
				    it != rows.end(); ++it)
				{
					if(! it->second.empty())
					{
						show_data(it->second.data());
						ptr_q->push_back(it->second.data_ptr());
					}
					else
					{
						std::cout << "skip empty mutation for row "
							  << it->second.row_name()
							  << std::endl;
					}
				}
				rows.clear();
			}
			else
			{
				const row_map::iterator it = rows.find(row);
				if(it != rows.end())
				{
					show_data(it->second.data());
					ptr_q->push_back(it->second.data_ptr());
					rows.erase(it);
				}
			}

			if(ptr_q->empty())
			{
				std::cout << "not data"
					  << (row.empty() ? "" : " for row \"")
					  << row << "\""
					  << std::endl;
			}
			else
			{
				uint64_t ts;
				try {
					ts = boost::lexical_cast<uint64_t>(timestamp);
				} catch(const std::exception &e) {
					ts = cur_micro_seconds();
				}

				boost::mutex::scoped_lock lock(m);
				++pending_ops;
				lock.unlock();

				manager->async_connect(ts_ep, boost::bind(on_connected,
									  tablet_name, ptr_q, ts,
									  _1, _2));
			}
		}
		else
		{
			command_error();
		}
	}

	boost::mutex::scoped_lock lock(m);
	while(pending_ops > 0)
	{
		while(! complete_info.empty())
		{
			std::cout << complete_info.front();
			complete_info.pop_front();
			--pending_ops;
		}
		if(pending_ops == 0)
			break;
		lock.unlock();
		std::cout << "wait pending ops ..." << std::endl;
		sleep(1);
		lock.lock();
	}

}
